package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo4Filter {

    Connection connection;

    /**
     * 创建链接
     */
    @Before
    public void init() {
        //1、创建hbase链接
        //创建hadoop配置对象
        Configuration configuration = HBaseConfiguration.create();
        //指定链接参数
        //指定zk地址，在zk中保存了hbase元数据
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        //建立链接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 列值过滤器，需要做全表扫描，性能比较低
     * <p>
     * 过滤器逻辑是运行在region server中的
     */
    @Test
    public void columnValueFilter() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));

        Scan scan = new Scan();
        /*
         * 取出文科所有学生
         *
         */
        //创建正则比较器
        RegexStringComparator regexStringComparator = new RegexStringComparator("文科*");

        //前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("理科"));

        //二进制比较器
        BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("文科一班"));

        //创建列值过滤器
        SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("info"),//列簇
                Bytes.toBytes("clazz"),//列名
                CompareFilter.CompareOp.EQUAL,//操作符
                binaryComparator);//比较强

        //增加过滤器
        scan.setFilter(columnValueFilter);

        //执行查询
        ResultScanner resultScanner = student.getScanner(scan);

        //解析数据
        for (Result result : resultScanner) {
            //去除rowkey
            String id = Bytes.toString(result.getRow());
            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }

    /**
     * 判断是否包含字符串
     */
    @Test
    public void subString() throws Exception {
        //获取表对象
        Table student = connection.getTable(TableName.valueOf("student"));

        Scan scan = new Scan();

        //比较器
        SubstringComparator comparator = new SubstringComparator("天");

        //列值过滤器
        SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("info"),
                Bytes.toBytes("name"),
                CompareFilter.CompareOp.EQUAL,
                comparator
        );

        scan.setFilter(columnValueFilter);

        //执行查询
        ResultScanner resultScanner = student.getScanner(scan);

        //解析数据
        for (Result result : resultScanner) {
            //去除rowkey
            String id = Bytes.toString(result.getRow());
            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }

    /**
     * 列名过滤器
     */
    @Test
    public void qualifierFilter() throws Exception {
        Table student = connection.getTable(TableName.valueOf("student"));

        //比较器
        SubstringComparator comparator = new SubstringComparator("a");
        //列名过滤器
        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, comparator);

        Scan scan = new Scan();

        scan.setFilter(qualifierFilter);

        //查询数据
        ResultScanner resultScanner = student.getScanner(scan);
        //解析数据
        for (Result result : resultScanner) {
            //去除rowkey
            String id = Bytes.toString(result.getRow());
            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }
    }


    /**
     * 行键过滤器，效率最高
     *
     */

    @Test
    public void rowFilter()throws Exception{
        Table student = connection.getTable(TableName.valueOf("student"));

        //前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("15001009"));

        //rowkey顾虑其
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        Scan scan = new Scan();
        //增加过滤器
        scan.setFilter(rowFilter);

        //执行查询
        ResultScanner resultScanner = student.getScanner(scan);

        for (Result result : resultScanner) {
            //去除rowkey
            String id = Bytes.toString(result.getRow());
            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


    @After
    public void close() {
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
